package com.baymax.test.scene

import androidx.collection.ArrayMap
import com.alibaba.fastjson.JSON
import com.alibaba.fastjson.serializer.JSONSerializer
import com.alibaba.fastjson.serializer.SerializeWriter
import com.badlogic.gdx.utils.Json
import com.baymax.test.scene.replace.StringTemplateUtilV2
import com.baymax.test.scene.url.IWrapperUrl
import com.baymax.test.scene.url.IWrapperUrl.WrapperUrl.OriginUrl
import com.baymax.test.scene.url.IWrapperUrl.WrapperUrl.VodUrl
import com.google.gson.Gson
import com.google.gson.JsonObject
import com.google.gson.JsonParser
import com.google.gson.internal.Streams
import com.google.gson.stream.JsonWriter
import com.jakewharton.rxrelay2.BehaviorRelay
import io.reactivex.*
import io.reactivex.Observable
import io.reactivex.Observer
import io.reactivex.disposables.Disposable
import io.reactivex.disposables.Disposables
import io.reactivex.observers.DisposableObserver
import io.reactivex.observers.DisposableSingleObserver
import io.reactivex.schedulers.Schedulers
import io.reactivex.subjects.BehaviorSubject
import kotlinx.coroutines.*
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.channels.consumeEach
import kotlinx.coroutines.flow.catch
import kotlinx.coroutines.flow.consumeAsFlow
import kotlinx.coroutines.flow.onCompletion
import kotlinx.coroutines.flow.onEach
import kotlinx.coroutines.rx2.asFlow
import kotlinx.coroutines.rx2.await
import kotlinx.coroutines.rx2.awaitLast
import kotlinx.coroutines.rx2.collect
import okhttp3.HttpUrl
import okhttp3.HttpUrl.Companion.toHttpUrlOrNull
import okio.Buffer
import okio.ByteString
import okio.ByteString.Companion.decodeBase64
import okio.ForwardingSource
import okio.Okio
import okio.buffer
import okio.sink
import okio.source
import org.json.JSONObject
import org.junit.Test
import java.io.File
import java.io.IOException
import java.io.StringWriter
import java.lang.AssertionError
import java.lang.Runnable
import java.lang.ref.WeakReference
import java.net.URLDecoder
import java.net.URLEncoder
import java.time.Instant
import java.util.*
import java.util.concurrent.CountDownLatch
import java.util.concurrent.Executors
import java.util.concurrent.TimeUnit
import java.util.concurrent.atomic.AtomicInteger
import java.util.concurrent.locks.ReentrantLock
import java.util.regex.Matcher
import java.util.regex.Pattern
import kotlin.concurrent.thread
import kotlin.random.Random
import kotlin.random.nextInt
import kotlin.time.*
import kotlin.time.Duration.Companion.seconds

/**
 * Example local unit test, which will execute on the development machine (host).
 *
 * See [testing documentation](http://d.android.com/tools/testing).
 */
fun logAny(o: Any?) {
    println("@${Thread.currentThread().name} -> $o")
}

class ExampleUnitTest {


    @Test
    fun addition_isCorrect() {
        val price = """
${'$'}_9.99
"""
        println(price)
        println(Dispatchers.Main.asExecutor())

    }

    @Test
    fun `testFastJson`() {
        println("""{author:\$\{author}""")
        val m = WLMaterial().apply {
            aid = "1"
            mid = "2"
            event_content = "event_content"
            file_name = "file_name"
            file_path = "file_path"
            file_md5 = "file_md5"
            file_size = "file_size"
            font_stype = "font_stype"
            click = listOf("click1", "click2")
            setPos("setPos")
        }
        println(JSON.toJSON(m))

    }

    val list = mutableListOf(15, 7, 5, 4, 9, 0, 2)

    @Test
    fun `test list compare`() {
        list.sortWith(Comparator { i, j ->
            i - j
        })

        println(list)
    }


    val worker = Schedulers.io().createWorker()

    val task: Runnable = Runnable {
        log("task start")
        Thread.sleep(1000)
        log("task end")

    }

    @Test
    fun `test schedules run if dispose`() {
        val dis = worker.schedule(task)
        runBlocking {
            repeat(3) {
                delay(400)
                worker.schedule(task)
                log("task -> ${dis.isDisposed}")
            }
            delay(3000)
        }

    }

    lateinit var lateD: Disposable

    @Test
    fun `test single success is dispose`() {
        runBlocking {
            val c = CountDownLatch(1)
            lateD = Single.timer(1, TimeUnit.SECONDS, singleTestSchedulers)
                .observeOn(singleTestSchedulers)
                .doOnSuccess {
                    log("test single success is dispose: 1 ${lateD.isDisposed}")
                    task.run()
                    log("test single success is dispose: 2 ${lateD.isDisposed}")
                }
                .subscribe { i ->
                    log("subscribe1 ${lateD.isDisposed}")
                    task.run()
                    log("subscribe2 ${lateD.isDisposed}")
                    //ss c.countDown()
                }
            c.await()
        }

    }

    val fix = Executors.newFixedThreadPool(2)
    val singleTestSchedulers = Schedulers.from(fix)

    @Test
    fun `test single success is dispose and wait in schedules`() {
        runBlocking {
            val c = CountDownLatch(10)
            val start = System.currentTimeMillis()
            repeat(10) {
                val sub = object : DisposableSingleObserver<Long>() {
                    override fun onSuccess(t: Long) {
                        log(fix.toString())
                        log(singleTestSchedulers.toString())
                        log("subscribe1 ${this}")
                        task.run()
                        log("subscribe2 ${this.isDisposed}")
                        c.countDown()
                    }

                    override fun onError(e: Throwable) {
                        TODO("Not yet implemented")
                    }
                }
                Single.timer(1, TimeUnit.SECONDS, singleTestSchedulers)
                    .observeOn(singleTestSchedulers)
                    .doOnSubscribe { log("single start") }
                    .doOnSuccess {
                        log("test single success is dispose: 1 ${sub.isDisposed}")
                        log("test single success is dispose: 2 ${sub.isDisposed}")
                    }
                    .subscribe(sub)

            }
            c.await()
            val end = System.currentTimeMillis()
            log("test single success is dispose and wait in schedules: cost ${start - end}ms")
        }

    }


    @Test
    fun `test completeable from run is disposoble`() {
        runBlocking {
            val l = withContext(Dispatchers.IO) {
                buildList {
                    repeat(10) {
                        add(
                            Completable.fromRunnable(task)
                                .subscribeOn(Schedulers.io())
                                .subscribe()
                        )

                    }
                }
            }
            repeat(6) {
                log(l.joinToString { it.isDisposed.toString() })
                delay(500)
            }
            log(l.joinToString { it.isDisposed.toString() })
        }
    }

    @Test
    fun `test observable sub`() {
        runBlocking {
            lateD = Observable.just("").subscribeOn(Schedulers.io())
                .observeOn(Schedulers.io())
                .subscribe({
                    log(lateD.isDisposed.toString())
                    task.run()
                    log(lateD.isDisposed.toString())
                }, {}, { log("complete ${lateD.isDisposed}") })
            delay(3000)
        }
    }


    @Test
    fun `test http url support rtp`() {

        val url = HttpUrl.Builder()
            .scheme("rtp")
            .host("121.121.12.12")
            .addPathSegment("cctv1")
            .build()
        println(url)
        assert(url != null)
    }


    data class TimeNode(val stamp: Long)

    @Test
    fun `test list partion`() {
        fun nextRandom() = (Math.random() * 15000).toLong() + 1000

        val l = buildList {
            repeat(10) {
                if (it % 2 == 0) {
                    add(TimeNode(System.currentTimeMillis() + nextRandom()))
                } else {
                    add(TimeNode(System.currentTimeMillis() - nextRandom()))
                }
            }
        }
        println(l)
        println(l.partition { it.stamp >= System.currentTimeMillis() })
    }


    @Test
    fun `test job when complete`() {
        val j = Job()
        thread {
            Thread.sleep(1000)
            j.complete()
        }
        runBlocking {
            log("do start")
            j.join()
            log("do next")
        }
    }


    val sub = BehaviorRelay.createDefault(-1)
    val pre = BehaviorSubject.createDefault(false)

    @Test
    fun `test job as comp`() {
        runBlocking {
            val d = sub.compose(composeWait2UserVisible()).subscribe {
                log("sub value $it")
            }
            launch {
                delay(2000)
                pre.onNext(true)
            }
            log("test job as comp: ${d.isDisposed}")
            repeat(10) {
                sub.accept(it)
                delay(600)
            }
            log("test job as comp: ${d.isDisposed}")
            d.dispose()
        }

    }

    fun <U> composeWait2UserVisible(): ObservableTransformer<U, U> {
        return ObservableTransformer { u ->
            Completable.fromObservable(pre.takeUntil { it }).andThen(u)
        }
    }

    private val serialWorker = Schedulers.io().createWorker()

    @OptIn(ExperimentalTime::class)
    @Test
    fun `test job as worker`() {
        val c = CountDownLatch(10)

        runBlocking {
            val d = measureTime {
                repeat(10) {
                    val i = (Math.random() * 10).toLong()
                    log("start task $i.")
                    serialWorker.schedule(
                        {
                            log("start task $i ,seconds ...")
                            c.countDown()
                        }, i, TimeUnit.SECONDS
                    )
                }
                c.await()
            }
            log("cost $d")

        }

    }

    private fun requestError(i: Int): Observable<String> {
        return Observable.timer(1, TimeUnit.SECONDS)
            .map {
                if (i == 2) {
                    error("test error")
                } else i.toString()
            }
    }

    val ob = object : DisposableObserver<String>() {
        override fun onNext(t: String) {
            log(t)
        }

        override fun onError(e: Throwable) {
            log("onError")
            e.printStackTrace()
        }

        override fun onComplete() {
            log("onComplete")
        }
    }

    @Test
    fun `test switch on err`() {
        val d = sub.flatMap { requestError(it) }
            .onErrorReturn { it.message }
            .subscribeOn(Schedulers.io())
            .subscribe({ log(it) }, { it.printStackTrace() }, { log("onComplete") })
        log("sub ${d.isDisposed}")
        runBlocking {
            repeat(3) {
                sub.accept(it)
                delay(800)
            }
            log("sub ${d.isDisposed}")
            delay(5000)
        }
        log("sub ${d.isDisposed}")


    }

    val detailState = BehaviorRelay.create<Int>()

    @Test
    fun `test delay func`() {
        runBlocking {
            launch {
                log("detail start")
                delay(2000)
                log("detail ok")
                detailState.accept(1)
            }
            val detail = detailState.takeUntil { it > 0 }.ignoreElements()

            val pay = Observable.create<Int> { e ->
                launch {
                    detailState.accept(-1)
                    delay(600)
                    log("pay state ok")
                    e.onNext(1)
                    e.onComplete()
                }
            }

            log("start")
            val i = pay.delay { i -> detail.toObservable<Int>() }
                .doOnNext { log("end") }
                .awaitLast()

            log("com")
        }

    }

    fun createPay(): Observable<Int> {
        log("pay state start1 ...")
        return Observable.create<Int> { e ->
            log("pay state start2 ...")
            GlobalScope.launch {
                detailState.accept(-1)
                delay(600)
                log("pay state ok")
                e.onNext(1)
                e.onComplete()
            }
        }

    }

    @Test
    fun `test complete and then`() {
        runBlocking {
            launch {
                repeat(10) {
                    delay(600)
                    log("push $it")
                    detailState.accept(it)
                }
            }

            val pp = detailState.takeUntil { it == 6 }
                .ignoreElements()
                .andThen(Completable.fromRunnable { log("do next ") })
                .andThen(createPay())
                .awaitLast()
            log("pay state $pp")
        }
    }

    @Test
    fun `test finnaly`() {
        val l = ReentrantLock()

        runBlocking {
            kotlin.runCatching {
                val s = Single.create<Int> {
                    launch {
                        delay(2000)
                        it.onSuccess(1)
                    }
                }


                try {
                    l.lock()
                    log("s run")
                    s.timeout(1, TimeUnit.SECONDS).blockingGet()
                } catch (e: Exception) {
                    log("e:$e")
                } finally {
                    l.unlock()
                    log("finally")
                }
            }.onFailure {
                log("onFailure")
            }

        }


    }


    @Test
    fun `test zip`() {
        runBlocking {
            val ob1 = Observable.create<Int> {
                log("start 1")
                it.onNext(1)
            }
            val ob2 = Observable.create<Int> {
                log("start 2")
                it.onNext(2)
            }
            Observable.zip(ob1, ob2) { a, b -> a + b }
                .subscribeOn(Schedulers.io())
                .map { log("sub $it") }
                .blockingFirst()
        }

    }

    @Test
    fun `subOnError`() {
        runBlocking {
            val d = Observable.create<Int> {
                it.onNext(1)
                it.onNext(2)
                it.onNext(3)
            }.subscribe({
                log("it $it")
                if (it == 2) {
                    1 / 0
                }
            }, {
                log("error")

                1 / 0

            })
            delay(3000)
        }
    }

    @Test
    fun `test complete with value`() {
        val cj = CompletableDeferred<Int>(null)
        log("invokeOnCompletion2222 $cj")
        log("start")
        cj.invokeOnCompletion {
            log("invokeOnCompletion")
        }
        runBlocking {

            launch {
                kotlin.runCatching {
                    delay(1000)
                    cj.complete(2)
                    log("end $cj")
                    log("end ${cj.await()}")
                }.onFailure {
                    log("invokeOnCompletion2222 $it")
                }
            }
        }
    }

    @Test
    fun `test collection fold`() {
        runBlocking {
            val l = listOf(1, 2, 3, 4, 5).fold(mutableListOf<String>()) { acc, i ->
                acc.apply {
                    acc.add("s:${i}")
                }
            }
            log("list: $l")

            val s = listOf(1, 2, 3, 4, 5).associateBy { it.toString() }
            listOf(1, 2, 3, 4, 5).mapIndexed { index, i -> }
        }


    }

    @Test
    fun `test cotact with`() {
        runBlocking {
            listOf(100, 300, 800).fold(Completable.complete()) { acc, t ->
                return@fold acc.concatWith(
                    Completable.timer(t.toLong(), TimeUnit.MILLISECONDS).doOnComplete {
                        log("doOnComplete1 $t")
                    }).doOnComplete {
                    log("doOnComplete2 $t")
                }

            }.await()


        }


    }

    val subject = BehaviorRelay.create<Completable>()

    var current: CompletableEmitter? = null

    @Test
    fun `test setDisposable`() {

        val create = { index: Int ->
            Completable.create { e ->
                log("create  $index")
                current = e

                e.setDisposable(Completable.timer(600, TimeUnit.MILLISECONDS)
                    .doOnDispose { e.onComplete() }
                    .subscribe {
                        log("timer end $index")
                        e.onComplete()
                    }
                )
            }.doOnComplete {
                log("timer complete $index")
            }.doOnDispose {
                log("timer cancel $index")
            }
        }

        fun post(index: Int) {
            subject.accept(create(index))
        }

        runBlocking {
            val j1 = subject.concatMapCompletable { it }.subscribe()
            //remain last
//           val j1 = Completable.concat(subject.toFlowable(BackpressureStrategy.LATEST)).subscribe()
//            val j1 = Completable.concat(subject.toFlowable(BackpressureStrategy.DROP)).subscribe()
            val j2 = launch {
                repeat(10) {
                    post(it)
                }

                delay(1000)
                log("d1: ${current} $j1")
                delay(1000)
                log("d2 1: ${current}  $j1")
                current?.onComplete()
                log("d2 2: ${current}  $j1")
                delay(1000)
                log("d3: ${current}  $j1")
                delay(1000)
            }
            j2.join()
            delay(8000)
        }
    }

    @Test
    fun `test compare`() {
        log("2022-08-31 11:16:00".compareTo("2022-08-31 15:03:06").toString())
        log(TimeZone.getDefault().rawOffset.toString())
    }

    @Test
    fun `testOKIO`() {
        val p = File("../test/")
        println(p.absolutePath)
        p.mkdirs()
        val f = File(p, "a.txt").sink().buffer()
        f.writeInt(1)
        f.writeInt(2)
        f.close()
        val s = File(p, "a.txt").source().buffer()
        val i = s.readInt()
        println(i)
        val i2 = s.readInt()
        println(i2)
    }

    @Test
    fun `testOKIO2`() {
        val source = File("D:\\dev\\android\\projects\\bay\\2022-12\\03-江西合并\\icntv.apk")
        val f = object : ForwardingSource(source.source()) {
            val total = source.length()
            var totalBytesRead = 0L

            override fun read(sink: Buffer, byteCount: Long): Long {
                val bytesRead = super.read(sink, byteCount)
                totalBytesRead += if (bytesRead == -1L) 0L else bytesRead
                println("progress $bytesRead change $totalBytesRead , $total , ${totalBytesRead * 100 / total}")
                return bytesRead
            }
        }.buffer()
        val p = File("../test/")
        val w = File(p, "a.apk").sink().buffer()
        w.writeAll(f)
        w.close()

    }


    @Test
    fun `testC`() {
        val d = Completable.create { e ->
            val d = Observable.timer(3, TimeUnit.SECONDS)
                .map { emptyList<Int>() }
                .flatMapIterable { r -> r }
                .firstOrError()
                .subscribe({
                    println("Observable next $it")
                    e.onComplete()
                }, {
                    println("Observable error $it")
                    e.onComplete()
                })
            e.setDisposable(d)
        }.subscribe({ println("Completable complete") })
        runBlocking {
            delay(2000)
            d.dispose()
            println("dispose")
            delay(3000)
        }

    }

    @Test
    fun testStr() {
        println(Arrays.toString(arrayOf("get", "CDN")))
    }

    @OptIn(ExperimentalTime::class)
    @Test
    fun testArray() {
        val TEST_SIZE = 10_000

        val map: MutableMap<Int, String> = ArrayMap(16)
        var record: TimedValue<String?> = TimedValue(null, Duration.ZERO)

        repeat(TEST_SIZE) {
            map.put(it, it.toString())
            if (it > 5) {
                val size = map.size
                val r = Random.nextInt(size)
                var t = measureTimedValue {
                    map.get(r)
                }
                log("m1:$t in $size")
                if (t.duration > record.duration) {
                    record = t
                }
            }

        }
        println("======================= 1$record")
        record = TimedValue(null, Duration.ZERO)
        val map2: MutableMap<Int, String> = HashMap()

        repeat(TEST_SIZE) {
            map2.put(it, it.toString())
            if (it > 5) {
                val size = map2.size
                val r = Random.nextInt(size)
                var t = measureTimedValue {
                    map.get(r)
                }
                log("m2:$t in $size")
                if (t.duration > record.duration) {
                    record = t
                }
            }

        }
        println("======================= 2$record")
    }


    private var recordIds: MutableMap<Int, String> = ArrayMap(16)

    @Synchronized
    private fun getRecordIdMap(): MutableMap<Int, String> {
        if (recordIds.size >= 181 && recordIds is ArrayMap<*, *>) {
            recordIds = java.util.HashMap(recordIds)
        }
        return recordIds
    }

    @OptIn(ExperimentalTime::class)
    @Test
    fun testArrayReplace() {

        var record: TimedValue<String?> = TimedValue(null, Duration.ZERO)
        repeat(10000) {
            val recordIdMap = getRecordIdMap()
            recordIdMap.put(it, it.toString())
            if (it > 5) {
                val size = recordIdMap.size
                val r = Random.nextInt(size)
                var t = measureTimedValue {
                    recordIdMap.get(r)
                }
                log("testArrayReplace:$t in $size for ${recordIdMap.javaClass.simpleName}")
                if (t.duration > record.duration) {
                    record = t
                }
            }

        }
        println("record$record")
    }

    @Test
    fun testMath() {
        println(Math.ceil(0.901562))
        println(Math.ceil(1.0123))
    }

    @Test
    fun testHttpUrl() {
        val m = mapOf("hellp" to "world", "user" to null, "name" to null, "ww" to "")
        val url = "https://www.oschina.net/".toHttpUrlOrNull()!!.newBuilder().apply {
            m.forEach { (t, u) ->
                addEncodedQueryParameter(t, u)
            }
        }.build()

        log("http url $url")
    }

    @Test
    fun `testReduce`() {
        listOf(1, 2, 3, 4, 5).reduceIndexed { index, acc, i ->
            log("str:$index ,$acc ,$i")
            i
        }
    }

    val bean = Bean()

    @Test
    fun `testMedthodRef`() {
        logRun(this::testMedthodRef)
        logRun(this::testMedthodRef)
        logRun(bean::print)
        logRun(bean::print)
    }

    private fun logRun(runnable: Runnable) {
        println("[${Instant.now()}][${Thread.currentThread().name}]:$runnable")
    }

    private fun log(str: String) {
        println("[${Instant.now()}][${Thread.currentThread().name}]:$str")
    }

    @Test
    fun `testGson`() {
        val url =
            """https://ottlnyd-cloudpay.yys.mgtv.com:8980/#/CloudPay?=&appId=82123456&actionMode=2&pid=&psid=10012377664&contentName=%E4%BA%A4%E6%8D%A2%E4%BA%BA%E7%94%9F&cosInquiryInfo=[{"contentId":"10012377664","idType":"0","ppvId":"","productIdList":["P4558348146983714814","ydtv12Apro"],"productTags":""}]&productId=&itemId=&rightsId=&topicId=&topicName=&userTicket=&timeStamp=1678354911802&userId=21200105168459&source=null&fromChannel=&UserID=15142003671_220323110718@ott&UserToken=FLX60FLX603N7ej00Bmtm1HEttKKK2G9"""
        println(url)
        val j = com.alibaba.fastjson.JSONObject().apply {
            put("url", url)
        }
        val j1 = JsonObject().apply {
            addProperty("url", url)
        }
        println(j.toString())
        println(j1.toString())
        Matcher.quoteReplacement(url)

        println(url.jsonSerializer())
        println(url.jsonSerializer1())
        println(Matcher.quoteReplacement(url))
        println(Matcher.quoteReplacement("123"))
        val r = "jsonSerializer\\((\\w+?)\\)".toRegex().matchEntire("jsonSerializer(url)")

        println(r)
        println(r?.groupValues)
        val template = """
            {
            "url":"${'$'}{url}",
            "url1":"${'$'}{url}",
            "url2":"${'$'}{url2}",
            "time":"${'$'}{time}"
            }
        """.trimIndent()
        println(template)
        val templateJson: com.alibaba.fastjson.JSONObject =
            com.alibaba.fastjson.JSONObject.parse(template) as com.alibaba.fastjson.JSONObject
        val matcher = "\\$\\{(.+?)}".toRegex().findAll(template).toList()

        println(matcher)
        println(matcher.toList().joinToString { it.value })
        //println(StringTemplateUtilV2.replace(template , mutableMapOf("url" to "www.baidu.com" , "time" to System.currentTimeMillis().toString())))
        val p = mutableMapOf(
            "url" to "www.baidu.com",
            "t" to System.currentTimeMillis().toString(),
            "uid" to 123.toString(),
            "ott_id" to "ott_id1",
            "from" to "fromv",
            "widget_id" to "widget_id1",

            )
        val j2 = StringTemplateUtilV2.replace(TMPLATE, p)
        println(j2)
    }

    val TMPLATE = """
            
            {
                "uid": "${'$'}{uid}",
                "event_id": "expose",
                "ott_id": "${'$'}{ott_id}",
                "t": "${'$'}{t}",
                "utype": "${'$'}{utype}",
                "props": {
                    "bustype": "${'$'}{bustype}",
                    "from": "${'$'}{from}",
                    "widget_id": "${'$'}{widget_id}",
                    "url": ${'$'}{jsonSerializer(url)}
                }
            }
        
        """


    fun String.jsonSerializer(): String {
        val out = SerializeWriter()
        return out.use { JSONSerializer(out).write(this); out.toString() }

    }

    fun String.jsonSerializer1(): String {
        return try {
            val stringWriter = StringWriter()
            val jsonWriter = JsonWriter(stringWriter)
            jsonWriter.isLenient = true
            jsonWriter.value(this)
            stringWriter.toString()
        } catch (e: IOException) {
            throw AssertionError(e)
        }

    }

    val o = Bean()

    @Test
    fun `testRef`() {

        val w1 = WeakReference(o)
        val w2 = WeakReference(o)
        println(o)
        println(w1 == w2)
        println(o)
        println(w1)
        println(w2)
    }


    @Test
    fun `test url`() {
        val voda = VodUrl("123456", "abc")
        val vodb = VodUrl("123456", "abc")
        val vodc = VodUrl("123456", "abc1")
        assert(voda == vodb)
        assert(voda != vodc)
        val o1 = OriginUrl("url1")
        val o2 = OriginUrl("url1")
        assert(o1 == o2)
    }

    @Test
    fun `test jsnon url encode`() {
        val str = """{"programSeriesId":"1200219602","continueTime":935}"""

        val json = JsonParser().parse(str)
        println("json $json")

        val en1 = URLEncoder.encode(str)
        println("encode $en1")

        val json1 = JsonParser().parse(URLDecoder.decode(en1))
        println("json1 $json1")
        val en2 = ByteString.of(*str.toByteArray()).base64Url()
        println("encode $en2")
        val en3 = ByteString.of(*str.toByteArray()).base64()
        println("encode $en3")
        val b = "eyJwcm9ncmFtU2VyaWVzSWQiOiIxMjAwMjE5NjAyIiwiY29udGludWVUaW1lIjo5MzV9"
        println(String(b.decodeBase64()?.toByteArray() ?: byteArrayOf()))
    }

    @Test
    fun testJSON() {
        val js = com.alibaba.fastjson.JSONObject.parseObject("")
        println(js)
    }

    private val ce = object : CompletableOnSubscribe, Runnable {

        var emitter: CompletableEmitter? = null

        override fun subscribe(emitter: CompletableEmitter) {
            this.emitter = emitter
        }

        override fun run() {
            emitter?.onComplete()
        }
    }

    @Test
    fun `test complete`() {
        val c = Completable.create(ce).cache()
        c.subscribe {
            println("sub1")

        }
        c.subscribe {
            println("sub1.1")
        }
        c.andThen(Completable.fromRunnable { println("Then2") }).subscribe { println("sub2") }

        ce.run()
        runBlocking {
            delay(100)
            c.subscribe {
                println("sub4.1")
            }
            c.andThen(Completable.fromRunnable { println("Then3") })
                .subscribe { println("sub3") }
            delay(1000)
        }
    }

    private val relay = BehaviorRelay.createDefault("hello")

    @Test
    fun testrxjava() {
        var disposable: Disposable = Disposables.empty()
        relay
            .map {
                if (it == "error") {
                    throw Throwable("test error")
                }
                it
            }
            .map { it + "_suffix" }


            .onErrorResumeNext(relay)
            .doOnNext { println("each $it") }
            .subscribe(object : Observer<String> {
                override fun onSubscribe(d: Disposable) {
                    println("sub $d")
                    disposable = d
                }

                override fun onError(e: Throwable) {
                    println("onError $e")
                }

                override fun onComplete() {
                    println("complete")
                }

                override fun onNext(t: String) {
                    println("next $t")
                }

            })

        println("handle $disposable")

        runBlocking {
            delay(300)
            relay.accept("world")
            delay(1000)
            relay.accept("error")
            delay(100)
            relay.accept("world1")
            delay(800)
            println(" ------------- ${disposable.isDisposed}")
        }


    }

    val threadSend = Dispatchers.IO.limitedParallelism(1)
    val threadReceive = Dispatchers.IO.limitedParallelism(1)
    val threadHttp = Dispatchers.IO.limitedParallelism(1).asExecutor()

    val send = CoroutineScope(threadSend)
    val receive = CoroutineScope(threadReceive + SupervisorJob())
    val r = Random(10000)

    fun postHttp(id: Int): Deferred<Int> {
        val c = CompletableDeferred<Int>()
        threadHttp.execute {
            log("--> postHttp $id")
            Thread.sleep(r.nextInt(1, 3).seconds.inWholeMilliseconds)
            log("--> postHttp $id ok")
            c.complete(id)
        }
        return c
    }

    @Test
    fun testChannel() {
        val channel = Channel<Int>()
        val id = AtomicInteger(0)
        val sendJob = send.launch {
            while (true) {
                delay(100)
                val i = id.getAndIncrement()
                log("send $i")
                channel.send(i)
                if (i > 10) {
                    log("close ,,,,, ${channel.close()}")
                }

            }
        }

        val r = receive.launch {
            channel.consumeAsFlow()
                .onEach {
                    log("receive -> $it")
                    if (it % 3 == 0) {
                        val r = postHttp(it)
                        launch {
                            r.await()
                            log("receive ok -> $it")
                        }
                    }
                    log("receive complete -> $it")
                }
                .catch {
                    log("catch $it")
                }
                .onCompletion {
                    log("complete $it")
                }
                .collect {}
        }

        runBlocking {
            sendJob.join()
            r.join()
        }
    }

    val ob1 = Observable.create<String> {
        log("ob1")
        Thread.sleep(1000)
        it.onNext("ob1")
    }
    val ob2 = Observable.create<String> {
        log("ob2")
        Thread.sleep(1000)
        it.onNext("ob2")
    }

    @Test
    fun testDelaySub() {
        val c = SupervisorJob() + CoroutineExceptionHandler { coroutineContext, throwable ->  log(throwable.toString()) }
        runBlocking(c) {
            log("start")
                val d =launch {
                    ob1.delaySubscription(1, TimeUnit.SECONDS)
                        .concatMap { ob2 }
                        .asFlow().collect {
                            log(it)
                        }

                }

            launch {
                delay(300)
                d.cancel()
            }
        }
    }

}